package com.lzwk.tableSql.kafka.sink;

import com.lzwk.utils.MyKafkaUtil;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: CC
 * @Date: 2022/2/18 16:19
 */
public class EarthLog {
    public final String odsEarthLog(StreamTableEnvironment tableEnv, String topic, String groupId) {
        String sql = "create table source_stat_log(" +
                "            `pre_spm` VARCHAR," +
                "            `ref` VARCHAR," +
                "            `ts` TIMESTAMP(3)," +
                "            `ua` VARCHAR," +
                "            `url` VARCHAR," +
                "            `url_params` VARCHAR," +
                "            `spm` VARCHAR," +
                "            `tor_id` VARCHAR," +
                "            `sid` VARCHAR," +
                "            `uid` VARCHAR," +
                "            `rts` DOUBLE," +
                "            `evt` VARCHAR," +
                "            `evt_params` VARCHAR," +
                "            `ip` VARCHAR," +
                "            `tsk` VARCHAR," +
                "            `dt` VARCHAR," +
                "            `hr` VARCHAR" +
                ") with (" + MyKafkaUtil.getKafkaDDL(topic, groupId) + ")";
        tableEnv.executeSql(sql);
        return sql;
    }

    /**
     * 统一格式流写入
     * dwd_wk_log
     * dwd_wks_log
     * dwd_wka_log
     * dwd_lha_log
     * dwd_oc_log
     * dwd_lx_log
     * dwd_others_log
     */
    public final String dwdEarthLog(StreamTableEnvironment tableEnv,String topic,String groupId) {
        String sql = "create table source_stat_log(" +
                "        `uid` STRING," +
                "        `spm` STRING," +
                "        `evt` STRING," +
                "        `evt_params` VARCHAR," +
                "        `ts` TIMESTAMP," +
                "        `rts` DOUBLE," +
                "        `sid` STRING," +
                "        `tor_id` STRING," +
                "        `tsk` STRING," +
                "        `url_pat` STRING," +
                "        `url` STRING," +
                "        `url_params` VARCHAR," +
                "        `spm0` STRING," +
                "        `spm1` STRING," +
                "        `spm2` STRING," +
                "        `spm3` STRING," +
                "        `dt` STRING," +
                "        `hr` STRING" +
                ") with (" + MyKafkaUtil.getKafkaDDL(topic, groupId) + ")";
        tableEnv.executeSql(sql);
        return sql;
    }

    public final String DirtyEarthLog(StreamTableEnvironment tableEnv,String topic,String groupId) {
        String sql = "create table source_stat_log(" +
                "            `@timestamp` VARCHAR," +
                "            `@filepath` VARCHAR," +
                "            `log` VARCHAR" +
                ") with (" + MyKafkaUtil.getKafkaDDL(topic, groupId) + ")";
        tableEnv.executeSql(sql);
        return sql;
    }
}
